#ifndef _M_CHANNEL_H_
#define _M_CHANNEL_H_

#include "../mqthird/include/muduo/proto/dispatcher.h"
#include "../mqthird/include/muduo/proto/codec.h"
#include <muduo/net/TcpConnection.h>
#include "exchange.hpp"
#include "msgqueue.hpp"
#include "binding.hpp"
#include "message.hpp"
#include "virtualhost.hpp"
#include "routing.hpp"
#include "consumer.hpp"

#include "../mqcommon/threadpool.hpp"
#include "../mqcommon/mqProto.pb.h"
#include "../mqcommon/mqlink.pb.h"
#include "../mqcommon/logger.hpp"
#include "../mqcommon/helper.hpp"

namespace MyRabbitMQ
{

    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr = std::shared_ptr<protoMQ::openChannelRequest>;
    using closeChannelRequestPtr = std::shared_ptr<protoMQ::closeChannelRequest>;
    using declareExchangeRequestPtr = std::shared_ptr<protoMQ::declareExchangeRequest>;
    using deleteExchangeRequestPtr = std::shared_ptr<protoMQ::deleteExchangeRequest>;
    using declareQueueRequestPtr = std::shared_ptr<protoMQ::declareQueueRequest>;
    using deleteQueueRequestPtr = std::shared_ptr<protoMQ::deleteQueueRequest>;
    using bindRequestPtr = std::shared_ptr<protoMQ::bindRequest>;
    using unBindRequestPtr = std::shared_ptr<protoMQ::unBindRequest>;
    using basicPublishRequestPtr = std::shared_ptr<protoMQ::basicPublishRequest>;
    using basicAckRequestPtr = std::shared_ptr<protoMQ::basicAckRequest>;
    using queueConsumeRequestPtr = std::shared_ptr<protoMQ::queueConsumeRequest>;
    using queueCancelConsumeRequestPtr = std::shared_ptr<protoMQ::queueCancelConsumeRequest>;

    //信道类
    class Channel
    {
    public:
        using ptr = std::shared_ptr<Channel>;

        Channel(const std::string& cid
        ,const muduo::net::TcpConnectionPtr& conn
        ,const ProtobufCodecPtr& codec
        ,const threadpool::ptr& pool
        ,const VirtualHost::ptr& host
        ,const ConsumerManager::ptr& consumers)
        :m_id(cid)
        ,m_conn(conn)
        ,m_codec(codec)
        ,m_threadpool(pool)
        ,m_host(host)
        ,m_consumers(consumers)
        {
            LOG(logLevel::INFO) << "信道" << m_id << "创建" << std::endl;                  
        }

        ~Channel()
        {
            //如果消费者信息有效, 则将其从信道消费者管理句柄移除
            if(m_consumer != Consumer::ptr())
            {
                m_consumers->deleteConsumer(m_consumer->m_subscriptionQueue,m_consumer->m_name); 
                LOG(logLevel::INFO) << "信道" << m_id << "关闭" << std::endl;                  
            }
        }
        
        //声明交换机
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            //处理请求
            bool ret = m_host->declareExchange(req->exchange_name(),req->type(),req->isdurable(),req->isautodelete(),req->args()); 
            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),ret);
        }

        //删除交换机
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            //处理请求
            bool ret = m_host->deleteExchange(req->exchange_name());            
            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),ret);
        }

        //声明队列
        void declareQueue(const declareQueueRequestPtr& req)
        {
            //处理请求
            bool ret = m_host->declareMsgQueue(req->queue_name(),req->isunique()
            ,req->isdurable(),req->isautodelete(),req->args());

            if(!ret) return this->basicResponse(req->request_id(),req->channel_id(),false);

            m_consumers->declareQueueConsumer(req->queue_name());
            
            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

        //删除队列
        void deleteQueue(const deleteQueueRequestPtr& req)
        {
            bool ret = m_host->deleteMsgQueue(req->queue_name());

            if(!ret) return this->basicResponse(req->request_id(),req->channel_id(),false);

            m_consumers->deleteQueueConsumer(req->queue_name());

            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

        //声明绑定
        void bind(const bindRequestPtr& req)
        {
            bool ret = m_host->Bind(req->exchange_name(),req->queue_name(),req->binding_key());
            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

        //解除绑定
        void unBind(const unBindRequestPtr& req)
        {
            bool ret = m_host->unBind(req->exchange_name(),req->queue_name());
            //返回响应
            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

        //发布消息
        void basicPublish(const basicPublishRequestPtr& req)
        {
            //判断交换机是否存在
            if(!m_host->existsExchange(req->exchange_name())) 
                return this->basicResponse(req->request_id(),req->channel_id(),false);

            //交换机存在, 进行获取
            auto eptr = m_host->selectExchange(req->exchange_name());

            //进行交换路由, 获取交换机的所有绑定队列, 判定消息要发布到哪个队列中
            auto map = m_host->selectBindings(eptr->m_name);

            protoMQ::BaseProperties* properties = nullptr;
            std::string routing_key;

            //如果消息属性存在
            if(req->has_properties())
            {
                *properties =  req->properties();
                routing_key = properties->routing_key();
            }

            // 遍历所有绑定队列, 查看路由是否匹配
            for(auto& f : map)
            {
                if(Router::route(eptr->m_type,routing_key,f.second->m_bindingKey))
                {
                    //路由匹配成功, 将消息添加到该队列中
                    m_host->basicPublish(f.first,properties,req->body());

                    //这个消息添加到队列后, 就需要通知队列的订阅者, 让他们去消费消息--线程池负责完成
                    auto task = std::bind(&Channel::consume,this,f.first);
                    m_threadpool->push(task);
                    //正常来讲,服务器收到发布的消息,就会组织成为一个任务抛入线程池, 任务就是找个订阅客户端向他推送消息,这里绑定的是consume函数
                }
            }

            //返回响应
            return basicResponse(req->request_id(),req->channel_id(),true);
        }
        
        //确认消息
        void basicAck(const basicAckRequestPtr& req)
        {
            bool ret = m_host->basicAck(req->queue_name(),req->message_id());
            return this->basicResponse(req->request_id(),req->channel_id(),ret);
        }

        //消费者订阅队列
        void queueConsume(const queueConsumeRequestPtr& req)
        {
            //判断消费者订阅的队列是否存在
            if(!m_host->existsMsgQueue(req->queue_name()))
                return this->basicResponse(req->request_id(),req->channel_id(),false);

            auto func = std::bind(&Channel::consumeCallback,this,std::placeholders::_1,
            std::placeholders::_2,std::placeholders::_3);
            
            //队列存在, 所在队列添加消费者
            this->m_consumer = m_consumers->insertConsumer(req->consumer_tag(),req->queue_name(),
            req->isautoack(),func);

            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

        //消费者取消订阅队列
        void queueCancelConsume(const queueCancelConsumeRequestPtr& req)
        {
            //判断消费者订阅的队列是否存在
            if(!m_host->existsMsgQueue(req->queue_name()))
                return this->basicResponse(req->request_id(),req->channel_id(),false);

            //队列存在, 所在队列删除消费者
            m_consumers->deleteConsumer(req->queue_name(),req->consumer_tag());

            m_consumer.reset();

            return this->basicResponse(req->request_id(),req->channel_id(),true);
        }

    private:
        std::string m_id;                      //信道id
        muduo::net::TcpConnectionPtr m_conn;    //信道所属tcp连接
        ProtobufCodecPtr m_codec;                  //protobuf协议处理器, 处理序列化数据
        threadpool::ptr m_threadpool;                //线程工作池
        VirtualHost::ptr m_host;                //信道虚拟机管理句柄

        Consumer::ptr m_consumer;               //信道对应的消费者(这里的消费者是指订阅客户端, 所以该字段
                                                //不一定有效)
        ConsumerManager::ptr m_consumers;       //信道消费者管理句柄

        //发送响应
        void basicResponse(const std::string& rid, const std::string& cid, bool isOk)
        {
            //构造响应
            protoMQ::basicCommonResponse resp;
            resp.set_response_id(rid);
            resp.set_channel_id(cid);
            resp.set_ok(isOk);


            //发送响应,通过协议处理器, 进行序列化, 并通过连接发送
            m_codec->send(m_conn,resp);

        }

        //通知特定队列的订阅消费者, 让他们去消费信息
        void consume(const std::string& qname)
        {
            LOG(logLevel::TEST) << qname << std::endl;
            //判断队列是否存在, 不存在则报错
            auto qPtr = m_host->selectMsgQueue(qname);            
            if(qPtr.get() == nullptr) 
            {
                LOG(logLevel::ERROR) << "执行失败:未找到队列" << std::endl;
                return;
            }

            //队列存在, 从队列中取出一条信息
            auto mPtr = m_host->basicConsume(qname);
            if(mPtr.get() == nullptr) 
            {
                LOG(logLevel::ERROR) << "执行失败:未找到信息" << std::endl;
                return;
            }

            //取出一个消费者
            auto cPtr = this->m_consumers->getConsumer(qname);
            if(cPtr.get() == nullptr) 
            {
                LOG(logLevel::ERROR) << "执行失败:未找到消费者" << std::endl;
                return;
            }

            //消息推送操作, 调用订阅者的消息处理函数实现
            cPtr->m_callback(cPtr->m_name,&mPtr->payload().properties(),mPtr->payload().data());

            //如果订阅者设置了自动确认标志, 那么则不需要等待确认, 直接删除消息即可
            if(cPtr->m_autoAck) m_host->basicAck(qname,mPtr->payload().properties().id());

        }

        //创建消费者时, 基本回调函数
        //消息发给哪个消费者, 消息的属性, 消息的内容
        void consumeCallback(const std::string& tag, const protoMQ::BaseProperties *bp, const std::string &body) 
        {
            protoMQ::basicPushResponse resp;

            resp.set_consumer_tag(tag);
            resp.set_channel_id(this->m_id);
            resp.set_body(body);
            
            //如果bp不为nullptr
            if(bp)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_mode(bp->mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }

            //将消息发送出去
            this->m_codec->send(this->m_conn,resp);
        }
    };

    class ChannelManager
    {
    public:
        using ptr = std::shared_ptr<ChannelManager>;

        ChannelManager(){}

        //打开信道
        void openChannel(const std::string& cid
        ,const muduo::net::TcpConnectionPtr& conn
        ,const ProtobufCodecPtr& codec
        ,const threadpool::ptr& pool
        ,const VirtualHost::ptr& host
        ,const ConsumerManager::ptr& consumers)
        {
            auto it = std::make_shared<Channel>(cid, conn, codec, pool, host, consumers);
            {
                std::unique_lock<std::mutex>(m_mtx);

                //判断信道对象是否存在, 如果已经存在则返回
                if(m_channels.find(cid) != m_channels.end()) 
                {
                    return;
                }
                //不存在, 进行添加
                m_channels.insert(std::make_pair(cid,it));
            }
        }

        //关闭信道
        void closeChannel(const std::string& cid)
        {
            std::unique_lock<std::mutex>(m_mtx);
            
            //不存在则返回
            if(m_channels.find(cid) == m_channels.end())
            {
                return;
            }

            //存在, 进行删除
            m_channels.erase(cid);
        }

        //获取特定信道
        Channel::ptr getChannnel(const std::string& cid)
        {
            std::unique_lock<std::mutex>(m_mtx);
            //判断是否存在
            if(m_channels.find(cid) == m_channels.end())
            {
                return Channel::ptr();
            }

            return m_channels[cid];
        }

    private:
        std::mutex m_mtx;
        std::unordered_map<std::string, Channel::ptr> m_channels;
    };
}


#endif